package com.atguigu.flink.sql.query;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/26 0026 21:27
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink06_OverWindowSQL {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Long.valueOf(fields[1].trim()), Integer.valueOf(fields[2].trim()));
                        }

                );
        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //流转表
        Schema schema = Schema.newBuilder()
                .column("id" , "STRING")
                .column("vc" , "INT")
                .column("ts" ,"BIGINT")
                .columnByExpression("pt" , "PROCTIME()")
                .columnByExpression("et" , "TO_TIMESTAMP_LTZ(ts,3)")
                .watermark("et" , "et - INTERVAL '0' SECOND")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();
        tableEnv.createTemporaryView("t1" , table);

        // 基于行，上无边界下无边界(错误的)

        // 基于行，上无边界到当前行
        String  sql1 =
                " select id , vc, ts , "+
                        "   sum(vc) over(partition by id order by pt rows between UNBOUNDED preceding and current row ) svc " +
                        " from t1 "  ;

        // 基于行，上有边界到当前行
        String  sql2 =
                " select id , vc, ts , "+
                        "   sum(vc) over(partition by id order by pt rows between 2 preceding and current row ) svc " +
                        " from t1 "  ;


        //基于时间
        //上无边界， 下午边界(错误）
        // OVER RANGE FOLLOWING windows are not supported yet.
        //上无边界到当前时间
        String  sql3 =
                " select id , vc, ts , "+
                        "   sum(vc) over(partition by id order by et range between UNBOUNDED preceding and current row ) svc " +
                        " from t1 "  ;
        //上有边界到当前时间
        String  sql4 =
                " select id , vc, ts , "+
                        "   sum(vc) over(partition by id order by et range between interval '2' second preceding and current row ) svc " +
                        " from t1 "  ;


        //开窗，进行多次汇总
        /*
           SELECT order_id, order_time, amount,
              SUM(amount) OVER w AS sum_amount,
              AVG(amount) OVER w AS avg_amount
            FROM Orders
            WINDOW w AS (
              PARTITION BY product
              ORDER BY order_time
              RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW)
         */
        // Over Agg: Unsupported use of OVER windows. All aggregates must be computed on the same window. please re-check the over window statement.
        String  sql5 =
                " select id , vc, ts , "+
                        "   sum(vc) over w sumvc , " +
                        "   max(vc) over w maxvc ," +
                        "   min(vc) over w minvc " +
                        " from t1 " +
                        " window w as (partition by id order by et range between interval '2' second preceding and current row ) ";

        //使用窗口
        tableEnv.sqlQuery(sql5).execute().print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
